package com.manbu.live.id.generate.provider.service.impl;

import com.manbu.live.id.generate.provider.dto.mapper.IdGenerateMapper;
import com.manbu.live.id.generate.provider.dto.po.IdGeneratePO;
import com.manbu.live.id.generate.provider.service.IdGenerateService;
import com.manbu.live.id.generate.provider.service.bo.LocalSeqIdBO;
import com.manbu.live.id.generate.provider.service.bo.LocalUnSeqIdBO;
import jakarta.annotation.Resource;
import org.apache.ibatis.jdbc.RuntimeSqlException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author hanWang
 * @Title:
 * @Package
 * @Description:
 * @date 2024/1/2412:40 PM
 */

@Service
public class IdGenerateServiceImpl implements IdGenerateService, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(IdGenerateService.class);
    private static Map<Integer, LocalSeqIdBO> localSeqIdBOMap = new ConcurrentHashMap<Integer, LocalSeqIdBO>();
    private static Map<Integer, LocalUnSeqIdBO> localUnSeqIdBOMap = new ConcurrentHashMap<Integer, LocalUnSeqIdBO>();
    private static final Integer SEQ_ID = 1;
    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 16, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000),
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r);
                    thread.setName("id-generate-thread-" + ThreadLocalRandom.current().nextInt(1000));
                    return null;
                }
            });

    /**
     * 使用Semaphore 信号量来防止多线程并发 多次刷新id段
     */
    private static Map<Integer, Semaphore> semaphoreMap = new ConcurrentHashMap<>();

    /**
     * id段刷新优化 阈值为0.75 达到百分之75 执行异步任务创建 优化
     */
    private static final float UNDATE_RATE = 0.75f;

    @Resource
    private IdGenerateMapper mapper;

    /**
     * 无序id生成器
     *
     * @param id
     * @return
     */
    @Override
    public Long geUnSeqId(Integer id) {
        if (id == null) {
            LOGGER.error("[geUnSeqId] id is error,id is{}", id);
            return null;
        }
        LocalUnSeqIdBO localSeqIdBO = localUnSeqIdBOMap.get(id);
        if (localSeqIdBO == null) {
            LOGGER.error("[geUnSeqId] localSeqIdBO is null,id is{}", id);
            return null;
        }
        Long returnId = localSeqIdBO.getIdQueue().poll();
        if (returnId == null) {
            LOGGER.error("[geUnSeqId] returnId is null,id is{}", id);
        }

        /**
         * 异步 预执行刷新id段
         */
        this.refreshLocalUnSeqId(localSeqIdBO);
        return returnId;
    }


    /**
     * 有序id生成器
     *
     * @param id
     * @return
     */
    @Override
    public Long geSeqId(Integer id) {
        if (id == null) {
            LOGGER.error("[geSeqId] id is error,id is{}", id);
            return null;
        }
        LocalSeqIdBO localSeqIdBO = localSeqIdBOMap.get(id);
        if (localSeqIdBO == null) {
            LOGGER.error("[geSeqId] localSeqIdBO is null,id is{}", id);
            return null;
        }

        /**
         * 异步 预执行刷新id段
         */
        this.refreshLocalSeqId(localSeqIdBO);
        long andIncrement = localSeqIdBO.getCurrentNum().getAndIncrement();

        if (andIncrement> localSeqIdBO.getNextThreshold()) {
            LOGGER.error("[geSeqId] id  is over limit,id is{}", id);
            return null;
        }

        // 获取当前id 直增
        return andIncrement;
    }

    /**
     * 刷新本地有序的id段
     *
     * @param localSeqIdBO
     */
    private void refreshLocalSeqId(LocalSeqIdBO localSeqIdBO) {
        // 当前 id字段区间值
        long step = localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();
        /**
         * 使用Semaphore 信号量来防止多线程并发 多次刷新id段
         * 防止没扩容完成的时候过多线程进入到 if里面
         */
        if (localSeqIdBO.getCurrentNum().get() - localSeqIdBO.getCurrentStart() > step * UNDATE_RATE) {

            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is null ,id is{}", localSeqIdBO.getId());
                return;

            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                // 异步进行同步id字段的操作
                LOGGER.info("尝试开始进行同步id段的同步操作");
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            IdGeneratePO idGeneratePO = mapper.selectById(localSeqIdBO.getId());
                            tryUpdateMysqlRecord(idGeneratePO);
                            // 释放semaphore资源


                        } catch (Exception e) {
                            LOGGER.error("[refreshLocalSeqId] error is {}", e);
                        } finally {
                            semaphoreMap.get(localSeqIdBO.getId()).release();
                            LOGGER.info("有序id段同步完成，id is {}", localSeqIdBO.getId());
                        }

                    }
                });
            }


        }

    }

    /**
     * 刷新本地无序的id段
     *
     * @param localUnSeqIdBO
     */
    private void refreshLocalUnSeqId(LocalUnSeqIdBO localUnSeqIdBO) {
        long end = localUnSeqIdBO.getNextThreshold();

        long begin = localUnSeqIdBO.getCurrentStart();
        int remainSize = localUnSeqIdBO.getIdQueue().size();

        if ((end - begin) * 0.25 > remainSize) {
            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is null ,id is{}", localUnSeqIdBO.getId());
                return;

            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {

                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            /**
                             * 如果 前俩步执行完了 semaphoreMap.get(localUnSeqIdBO.getId()).release(); 没来得及释放 导致下一次生成id不可用
                             */
                            IdGeneratePO idGeneratePO = mapper.selectById(localUnSeqIdBO.getId());
                            tryUpdateMysqlRecord(idGeneratePO);
                            semaphoreMap.get(localUnSeqIdBO.getId()).release();
                            LOGGER.info("有序id段同步完成，id is {}", idGeneratePO.getId());
                        } catch (Exception e) {
                            LOGGER.error("[refreshLocalUnSeqId] error is {}", e);
                        } finally {
                            semaphoreMap.get(localUnSeqIdBO.getId()).release();
                            LOGGER.info("有序id段同步完成，id is {}", localUnSeqIdBO.getId());
                        }

                    }
                });


            }

        }
    }

    //spring 启动的时候 bean 初始化的时候会回调这里
    @Override
    public void afterPropertiesSet() throws Exception {

        List<IdGeneratePO> idGeneratePOList = mapper.selectAll();
        for (IdGeneratePO idGeneratePO : idGeneratePOList) {
            tryUpdateMysqlRecord(idGeneratePO);
            semaphoreMap.put(idGeneratePO.getId(), new Semaphore(1));
        }

    }

    /**
     * 更新mysql里面的分布式id的配置信息，占用对应id段
     *
     * @param idGeneratePO
     */
    private void tryUpdateMysqlRecord(IdGeneratePO idGeneratePO) {

        int updateResult = mapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());
        if (updateResult > 0) {
            localIdBoHandler(idGeneratePO);
            return;
        }
        for (int i = 0; i < 3; i++) {
            IdGeneratePO newIdGeneratePO = mapper.selectById(idGeneratePO.getId());

            updateResult = mapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());
            if (updateResult > 0) {
                localIdBoHandler(idGeneratePO);
//                LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
//                AtomicLong atomicLong = new AtomicLong(idGeneratePO.getCurrentStart());
//                localSeqIdBO.setId(idGeneratePO.getId() );
//                localSeqIdBO.setCurrentNum(atomicLong );
//                localSeqIdBO.setCurrentStart(idGeneratePO.getCurrentStart() );
//                localSeqIdBO.setNextThreshold(idGeneratePO.getNextThreshold()  );
//                localSeqIdBO.setCurrentNum(atomicLong );
//                localSeqIdBOMap.put(localSeqIdBO.getId(),localSeqIdBO);
                return;
            }

        }
        throw new RuntimeException("表id字段占用失败, 竞争过于激烈 id is :" + idGeneratePO.getId());
    }

    /**
     * 专门处理如何将id对象放入本地缓存中
     *
     * @param idGeneratePO
     */
    private void localIdBoHandler(IdGeneratePO idGeneratePO) {
        long currentStart = idGeneratePO.getCurrentStart();
        long nextThreshold = idGeneratePO.getNextThreshold();
        long currentNum = currentStart;
        // 判断数据库取出来的id配置是有序还是无序 1 有序 非 1 无序
        if (idGeneratePO.getIsSeq() == SEQ_ID) {
            // 有序存储
            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            AtomicLong atomicLong = new AtomicLong(currentStart);
            localSeqIdBO.setId(idGeneratePO.getId());
            localSeqIdBO.setCurrentStart(currentStart);
            localSeqIdBO.setNextThreshold(nextThreshold);
            localSeqIdBO.setCurrentNum(atomicLong);
            localSeqIdBOMap.put(localSeqIdBO.getId(), localSeqIdBO);
        } else {
            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setId(idGeneratePO.getId());

            localUnSeqIdBO.setCurrentStart(currentStart);
            localUnSeqIdBO.setNextThreshold(nextThreshold);
            long begin = idGeneratePO.getCurrentStart();
            long end = idGeneratePO.getNextThreshold();
            ConcurrentLinkedQueue idQueue = new ConcurrentLinkedQueue();
            ArrayList<Long> idList = new ArrayList<>();
            for (long i = begin; i < end; i++) {
                idList.add(i);
            }
            // 无序操作将有序集合打乱
            Collections.shuffle(idList);
            idQueue.addAll(idList);
            localUnSeqIdBO.setIdQueue(idQueue);
            localUnSeqIdBOMap.put(localUnSeqIdBO.getId(), localUnSeqIdBO);

        }
    }
}
